package com.upcoming.data.storm.kafka;

import com.upcoming.data.storm.util.Event;
import com.upcoming.data.storm.util.EventDeserializer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.Map;
import java.util.Properties;

/**
 * 作者：tianjiayuan
 * 创建时间：2018-05-23 14:24
 * 类描述：
 * 修改人：
 * 修改时间：
 */
public class KafkaSpout extends BaseRichSpout {

    private static final Logger logger = LoggerFactory.getLogger(KafkaSpout.class);

    private SpoutOutputCollector collector;

    private KafkaConsumer<String, Event> consumer;


    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("data"));
    }

    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        this.collector = spoutOutputCollector;
        if (null == consumer) {
            Properties props = new Properties();

            //设置borker地址
            props.put("bootstrap.servers", "192.168.1.210:9092,192.168.1.211:9092,192.168.1.215:9092,192.168.1.216:9092,192.168.1.217:9092");
            //设置消费者组Id
            String groupId = "DD0005";
            props.put("group.id", groupId);

            props.put("enable.auto.commit", "true");
            props.put("auto.commit.interval.ms", "1000");

            props.put("session.timeout.ms", "30000");

            props.put("key.deserializer", StringDeserializer.class.getName());
            props.put("value.deserializer", EventDeserializer.class.getName());

            consumer = new KafkaConsumer<String, Event>(props);

            consumer.subscribe(Arrays.asList(groupId));
        }
    }

    public void nextTuple() {
//        logger.info("nextTuple()");
        ConsumerRecords<String, Event> records = consumer.poll(100);
        if (!records.isEmpty()) {
            for (ConsumerRecord<String, Event> record : records) {
                try {
                    Event event = record.value();
                    if (event == null) {
                        logger.info(">>>>>>>>>>>>>>>>>>> event == null");
                        continue;
                    }
                    logger.info(">>>>>>>>>>>>>>>>>>>>" + event.getEventCode());
                    this.collector.emit(new Values(event));
                } catch (Exception e) {
                    logger.error(e.getMessage());
                }
            }
        }
    }

    @Override
    public void close() {
        super.close();
    }

    @Override
    public void fail(Object msgId) {
        super.fail(msgId);
    }
}
